package top.frsf.rpc.session;

import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import top.frsf.rpc.entity.ServiceInstance;
import top.frsf.rpc.server.NettyRpcClient;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 客户端 Channel 管理
 *
 * @author He Kuntao
 * @since 2025/7/15 10:09
 */
@Slf4j
public class ClientSessionManager implements DisposableBean {

    private final ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();

    private final NettyRpcClient rpcClient = new NettyRpcClient();

    /**
     * 获取或创建指定服务实例的 Channel
     *
     * @param instance 服务实例
     * @return Channel
     * @throws InterruptedException 中断异常
     */
    public Channel getOrCreateChannel(ServiceInstance instance) throws InterruptedException {
        String key = buildKey(instance);
        return channelMap.compute(key, (k, existingChannel) -> {
            try {
                if (existingChannel != null && existingChannel.isActive()) {
                    return existingChannel;
                }
                Channel newChannel = rpcClient.createConnection(instance.getIp(), instance.getPort());
                newChannel.closeFuture().addListener(f -> channelMap.remove(key));
                return newChannel;
            } catch (Exception e) {
                log.error("Connection failed to {}", key, e);
                return null;
            }
        });
    }

    /**
     * 强制关闭指定连接
     *
     * @param instance 服务实例
     */
    public void closeConnection(ServiceInstance instance) {
        Channel channel = channelMap.remove(buildKey(instance));
        if (channel != null) {
            channel.close();
        }
    }

    /**
     * 构建连接标识
     *
     * @param instance 服务实例
     * @return 连接标识
     */
    private String buildKey(ServiceInstance instance) {
        return instance.getIp() + ":" + instance.getPort();
    }

    @Override
    public void destroy() {
        channelMap.values().forEach(Channel::close);
        rpcClient.shutdown();
    }
}
